/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.sui.bigdata.flink.sql.launcher;

import avro.shaded.com.google.common.collect.Lists;
import com.sui.bigdata.flink.sql.core.util.PluginUtil;
import org.apache.commons.cli.BasicParser;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
import org.apache.commons.io.Charsets;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.util.Preconditions;
//import org.apache.flink.hadoop.shaded.com.google.common.base.Charsets;
//import org.apache.flink.hadoop.shaded.com.google.common.base.Preconditions;

import java.net.URLDecoder;
import java.net.URLEncoder;
import java.util.List;
import java.util.Map;


/**
 * Date: 2019/7/16
 * @author chenyong
 */
public class LauncherOptionParser {

    public static final String OPTION_MODE = "mode";

    public static final String OPTION_NAME = "name";

    public static final String OPTION_SQL = "sql";

    public static final String OPTION_FLINK_CONF_DIR = "flinkconf";

    public static final String OPTION_YARN_CONF_DIR = "yarnconf";

    public static final String OPTION_CORE_JAR_PATH = "coreJarPath";

    public static final String OPTION_ADDJAR = "addjar";

    public static final String OPTION_CONF_PROP = "confProp";

    public static final String OPTION_SAVE_POINT_PATH = "savePointPath";

    public static final String OPTION_ALLOW_NON_RESTORED_STATE = "allowNonRestoredState";

    public static final String OPTION_FLINK_JAR_PATH = "flinkJarPath";

    public static final String OPTION_UDF_JAR_PATH = "udfJarPath";

    public static final String OPTION_QUEUE = "queue";

    public static final String OPTION_META_TABLE= "metaTable";

    private Options options = new Options();

    private BasicParser parser = new BasicParser();

    private LauncherOptions properties = new LauncherOptions();

    public LauncherOptionParser(String sql, String name, String flinkconf, String udfJarPath, String yarnconf, String confProp, String metaTable) {
        options.addOption(OPTION_MODE, true, "Running mode");
        options.addOption(OPTION_SQL, true, "Job sql file");
        options.addOption(OPTION_NAME, true, "Job name");
        options.addOption(OPTION_FLINK_CONF_DIR, true, "Flink configuration directory");
        options.addOption(OPTION_CONF_PROP, true, "sql ref prop,eg specify event time");
        options.addOption(OPTION_YARN_CONF_DIR, true, "Yarn and hadoop configuration directory");
        options.addOption(OPTION_FLINK_JAR_PATH, true, "flink jar path for submit of perjob mode");
        options.addOption(OPTION_QUEUE, true, "flink runing yarn queue");
        options.addOption(OPTION_META_TABLE, true, "flink runing metaTable");
        options.addOption(OPTION_UDF_JAR_PATH, true, "flink streamsql udf");
        options.addOption(OPTION_CORE_JAR_PATH, true, "sql core path");
//        options.addOption(OPTION_ADDJAR, true, "sql ext jar,eg udf jar");
//        options.addOption(OPTION_SAVE_POINT_PATH, true, "Savepoint restore path");
//        options.addOption(OPTION_ALLOW_NON_RESTORED_STATE, true, "Flag indicating whether non restored state is allowed if the savepoint");

        try {


            if(sql.endsWith("txt") || sql.endsWith("sql")){
                sql = PluginUtil.readExecSql(sql);
            }else {
                sql = URLEncoder.encode(sql, Charsets.UTF_8.name());
            }

            String[] parameters = new String[]{"-sql",sql,"-name",name,"-flinkconf",flinkconf,"-udfJarPath",udfJarPath,"-yarnconf",yarnconf,"-confProp",confProp,"-metaTable",metaTable};
            CommandLine cl = parser.parse(options, parameters);

            String sqlfile = Preconditions.checkNotNull(cl.getOptionValue(OPTION_SQL),
                    "Must specify job file using option '" + OPTION_SQL + "'");
            properties.setSql(sqlfile);

            name = Preconditions.checkNotNull(cl.getOptionValue(OPTION_NAME));
            properties.setName(name);

            confProp = cl.getOptionValue(OPTION_CONF_PROP);
            if(StringUtils.isNotBlank(confProp)){
                properties.setConfProp(confProp);
            }
            flinkconf= cl.getOptionValue(OPTION_FLINK_CONF_DIR);
            if(StringUtils.isNotBlank(flinkconf)) {
                properties.setFlinkconf(flinkconf);
            }

            yarnconf = cl.getOptionValue(OPTION_YARN_CONF_DIR);
            if(StringUtils.isNotBlank(yarnconf)) {
                properties.setYarnconf(yarnconf);
            }


            udfJarPath = cl.getOptionValue(OPTION_UDF_JAR_PATH);
            if(StringUtils.isNotBlank(udfJarPath)){
                properties.setUdfJarPath(udfJarPath);
            }


            metaTable = cl.getOptionValue(OPTION_META_TABLE);
            if(StringUtils.isNotBlank(metaTable)){
                properties.setMetaTable(metaTable);
            }

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public LauncherOptions getLauncherOptions(){
        return properties;
    }

    public List<String> getProgramExeArgList() throws Exception {
        Map<String,Object> mapConf = PluginUtil.ObjectToMap(properties);
        List<String> args = Lists.newArrayList();
        for(Map.Entry<String, Object> one : mapConf.entrySet()){
            String key = one.getKey();
            if(OPTION_FLINK_CONF_DIR.equalsIgnoreCase(key)
                    || OPTION_YARN_CONF_DIR.equalsIgnoreCase(key)
                    || OPTION_FLINK_JAR_PATH.equalsIgnoreCase(key)
                    || OPTION_CORE_JAR_PATH.equalsIgnoreCase(key)
                    || OPTION_UDF_JAR_PATH.equalsIgnoreCase(key)){
                continue;
            }

            if(one.getValue() == null){
                continue;
            }

            args.add("-" + key);
            args.add(one.getValue().toString());
        }
        return args;
    }
}
